package cn.dmp.tags

import cn.dmp.util.Constans
import com.typesafe.config.ConfigFactory
import org.apache.commons.lang.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext}

/**
  * Created by Administrator on 2018/4/29.
  */
object Tags4Context2 {

  def main(args: Array[String]): Unit = {
    //加载路径参数
    val config = ConfigFactory.load()

    val conf = new SparkConf()
      .setMaster("local[*]").setAppName(s"${this.getClass.getSimpleName}")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf) //spark离线job的入口
    //读文件
    val sQLContext = new SQLContext(sc)
    val dataFrame = sQLContext.read.parquet(config.getString("InputPath"))

    //将appdic文件数据广播出去
    val appdict = sc.textFile(config.getString("appdict.path"))
    val appdicFilter = appdict.map(_.split("\t", -1)).filter(_.length >= 5).map({
      t => ((t(4), t(1)))
    }).collect().toMap
    val app = sc.broadcast(appdicFilter)
    val appDicBC = sc.broadcast(appdicFilter)

    //将stop文件广播出去
    val file = sc.textFile(config.getString("stopWords.path")).collect()
    val stopBC = sc.broadcast(file)

    val map = dataFrame.filter(Constans.havaUserIdLog).map({ row =>

      var tags = Map[String, Int]() //用来装处理好的数据

      //广告位类型和广告名称
      val adST = row.getAs[Int]("adspacetype")
      val adSTN = row.getAs[String]("adspacetypename")
      if (adST < 10) tags += "LC0" + adST -> 1 else tags += "LC" + adST -> 1
      if (StringUtils.isNotEmpty(adSTN)) "LN" + adSTN -> 1

      //APP类型
      val appid = row.getAs[String]("appid")
      var appName = row.getAs[String]("appname")
      if (StringUtils.isEmpty(appName)) {
        if (StringUtils.isNotEmpty(appid)){
          val appName=appDicBC.value.getOrElse(appid,appid)
          tags += "APP" + appName -> 1
        }else tags += "APP" + appName -> 1
      }else tags += "APP" + appName -> 1

      //渠道
      val adplat = row.getAs[Int]("adplatformproviderid")
      if (adplat > 0) tags += "CN" + adplat -> 1

      //设备
      val ispname = row.getAs[String]("ispname") //运营商
      val netname = row.getAs[String]("networkmannername") //网络类型
      val client = row.getAs[Int]("client") //操作系统

      client match {
        case 1 => tags+= "D00010001"-> 1
        case 2 => tags += "D00010002" -> 1
        case 3 => tags += "D00010003" -> 1
        case _ => tags += "D00010004" -> 1
      }

      netname match {
        case "WIFI" => tags += "D00020001" -> 1
        case "4G" => tags += "D00020002" -> 1
        case "3G" => tags += "D00020003" -> 1
        case "2G" => tags += "D00020004" -> 1
        case _ => tags += "D00020005" -> 1
      }

      ispname match {
        case "移动" => tags += "D00030001" -> 1
        case "联通" => tags += "D00030002" -> 1
        case "电信" => tags += "D00030003" -> 1
        case _ => tags += "D00030004" -> 1
      }

      //关键字
      val keyWords = row.getAs[String]("keywords")

      val filter = keyWords.split("\\|").filter(t => t.length >= 3 && t.length < 10 && !stopBC.value.contains(t))
        .foreach(t => tags += "K" + t -> 1)

      //地域标签
      val pname = row.getAs[String]("provincename")
      val cname = row.getAs[String]("cityname")
      if (StringUtils.isNotEmpty(pname)) tags += "ZP" + pname -> 1
      if (StringUtils.isNotEmpty(cname)) tags += "ZC" + pname -> 1

      (getUserId(row),tags.toList)

    }).reduceByKey({
      (list1,list2)=>(list1++list2).groupBy(t=>t._1).mapValues(f=>f.foldLeft(0)(_+_._2)).toList
        //.mapValues(t=>t.map(f=>f._2).sum).toList
    }).foreach(println)


    sc.stop()

  }

  def getUserId(row:Row):String={

    row match {
      case row if(row.getAs[String]("imei") !="") => "Imei:"+ row.getAs[String]("imei")
      case row if(row.getAs[String]("mac") !="") => "MAC:"+row.getAs[String]("mac")
      case row if(row.getAs[String]("idfa") !="") => "IDF:"+row.getAs[String]("idfa")
      case row if(row.getAs[String]("openudid") !="") => "OPE:"+row.getAs[String]("openudid")
      case row if(row.getAs[String]("androidid") !="") => "AND:"+row.getAs[String]("androidid")

      case row if(row.getAs[String]("imeimd5") !="") => "IMm5:"+row.getAs[String]("imeimd5")
      case row if(row.getAs[String]("macmd5") !="") => "MACm5:"+row.getAs[String]("macmd5")
      case row if(row.getAs[String]("idfamd5") !="") => "IDFm5:"+row.getAs[String]("idfamd5")
      case row if(row.getAs[String]("openudidmd5") !="") => "OPEm5:"+row.getAs[String]("openudidmd5")
      case row if(row.getAs[String]("androididmd5") !="") => "ANDm5:"+row.getAs[String]("androididmd5")

      case row if(row.getAs[String]("imeisha1") !="") => "IMs1:"+row.getAs[String]("imeisha1")
      case row if(row.getAs[String]("macsha1") !="") => "MACs1:"+row.getAs[String]("macsha1")
      case row if(row.getAs[String]("idfasha1") !="") => "IDFs1:"+row.getAs[String]("idfasha1")
      case row if(row.getAs[String]("openudidsha1") !="") => "OPEs1:"+row.getAs[String]("openudidsha1")
      case row if(row.getAs[String]("androididsha1") !="") => "ANDs1:"+row.getAs[String]("androididsha1")
    }

  }

}
